package com.xiguaji.tiktok.task.blogger;

import com.github.mustachejava.reflect.guards.MapGuard;
import com.gugee.tiktok.common.model.dal.blogger.*;
import com.gugee.tiktok.common.model.mysql.Blogger;
import com.gugee.tiktok.common.model.mysql.BloggerLog;
import com.gugee.tiktok.common.model.mysql.BloggerMain;
import com.gugee.tiktok.common.model.mysql.BloggerType;
import com.gugee.tiktok.data.common.msg.BloggerMessage;
import com.xiguaji.boot.toolkit.jdbc.JdbcTemplateHelper;
import com.xiguaji.tiktok.task.common.config.AppConfig;
import com.xiguaji.tiktok.task.common.job.TaskParam;
import com.xiguaji.tiktok.task.common.producer.AbstractProducerHandler;
import com.xxl.job.core.log.XxlJobLogger;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;

@Service
@Slf4j
public class BloggerService extends AbstractProducerHandler<BloggerMessage> {

    @Autowired
    BloggerMainRepository bloggerMainRepository;

    @Autowired
    BloggerLogRepository bloggerLogRepository;
    @Autowired
    BloggerRepository bloggerRepository;
    @Autowired
    @Qualifier("kafkaTemplateBlogger")
    private KafkaTemplate<Long, BloggerMessage> kafkaTemplateBlogger;
    @Autowired
    AppConfig appConfig;
    @Autowired
    BloggerTypeRepository bloggerTypeRepository;
    @Autowired
    AwemeRepository awemeRepository;
    DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyyMMdd");
    List<String> headRegion = new ArrayList<>();


    @PostConstruct
    public void postConstruct(){
        List<Map<String,Object>> result = JdbcTemplateHelper.select("SELECT code FROM `dic_region` where type = 1");

        result.forEach(x -> headRegion.add(x.get("code").toString()));
    }

    private List<BloggerType> getAllBloggerTypes(){
        List<BloggerType> bloggerTypes = new ArrayList<>();
        BloggerType bloggerType = new BloggerType();
        bloggerType.setId(0);
        bloggerType.setName("ALL");
        bloggerTypes.add(bloggerType);
        bloggerTypes.addAll(bloggerTypeRepository.getBloggerTypeByIsDisplay(true));
        return bloggerTypes;
    }
    public String getBloggerType(long oUid){
        return bloggerMainRepository.findTypeByoUid(oUid);
    }

    public List<BloggerMain> getBloggerRanting(String bloggerType, int limit){
        return bloggerMainRepository.getBloggersByBloggerTypeSortDesc(bloggerType,limit);
    }

    public List<BloggerMain> getBloggerEnterpriseRanting(String bloggerType, int limit){
        return bloggerMainRepository.getBloggersByEnterpriseVerify(bloggerType,limit);
    }
    public List<BloggerMain> getBloggerRegionRanting(String bloggerType, int limit){
        return bloggerMainRepository.getBloggersByRegion(bloggerType,limit);
    }

//    public List<BloggerMain> getBloggers(int id,int count){
//        return bloggerMainRepository.findBloggersByIdAndCount(id,count);
//    }

    public BloggerMain getBloggerByOUid(long oUid){
        val bloggerMainOptional = bloggerMainRepository.findByoUid(oUid);

        if(bloggerMainOptional.isPresent()){
            return bloggerMainOptional.get();
        }
        return null;
    }

    public Optional<BloggerLog> getBloggerLogByLogType(int logType){
        return bloggerLogRepository.findByLogType(logType);
    }

    public void saveBloggerMain(BloggerMain bloggerMain){
        bloggerMainRepository.save(bloggerMain);
    }

    public void saveBloggerLog(BloggerLog bloggerLog){
        bloggerLogRepository.save(bloggerLog);
    }

    @Override
    public List<BloggerMessage> findMessageObject(BloggerLog bloggerLog, TaskParam taskParam) {
        //两种模式的切换
//        if(appConfig.getIntervalHours()>0){
                    // return bloggerMainRepository.findBloggerMessagesForFollowerCountAndCount(appConfig.getSysCollectFollowerCount(),0,taskParam.getRecords(),appConfig.getIntervalHours());
//        }else {
//            //该模式下对页数的记载会变成无意义
//            return bloggerMainRepository.findBloggerMessagesForFollowerCountAndCount(appConfig.getSysCollectFollowerCount(), (bloggerLog.getCurrentBloggerId()-1)*taskParam.getRecords(), taskParam.getRecords());
//        }
        if(taskParam.getTopic().equals("blogger_fans_portrait")){
            return bloggerMainRepository.findBloggerMessagesForFollowerCountAndFansPortrait(appConfig.getSysCollectFollowerCount(),0,taskParam.getRecords(),appConfig.getIntervalHoursFansPortrait(),headRegion);
        }
        return bloggerMainRepository.findBloggerMessagesForFollowerCountAndCount(appConfig.getSysCollectFollowerCount(),0,taskParam.getRecords(),appConfig.getIntervalHours(),headRegion);

    }

    @Override
    public void sendMsg(List<BloggerMessage> messages,String topic) {
        messages.forEach(msg->{
            msg.setRetryTimes(appConfig.getRetryBloggerTimes());
            msg.setRetryLimitTimes(appConfig.getRetryBloggerTimes());
            kafkaTemplateBlogger.send(topic,msg);
        });
    }

    public void sendMsgWithAllTypes(TaskParam taskParam){
        LocalDateTime localDateTime =  LocalDateTime.now().minusDays(appConfig.getLatestDays()+1);
        int condition = Integer.parseInt(localDateTime.format(formatter));
        Set<Long> msgSended=new HashSet<>(taskParam.getRecords()*60);
        getAllBloggerTypes().forEach(b->{
            //从播主排行版获取top
            List<BloggerMessage> messages=bloggerMainRepository.findBloggerMessagesByBloggerTypeSortDesc(appConfig.getSysCollectFollowerCount(),b.getName().equals("ALL")?"":b.getName(), taskParam.getRecords(),appConfig.getIntervalHours());
            //从视频排行版获取top
            List<BloggerMessage> messages2=awemeRepository.findBloggerByTypeSortDescFromAweme(condition,b.getName().equals("ALL")?"":b.getName(), taskParam.getRecords(),appConfig.getIntervalHours());


            messages.forEach(msg->{
               if(!msgSended.contains(msg.getUid())) {
                   kafkaTemplateBlogger.send(taskParam.getTopic(), msg);
                   msgSended.add(msg.getUid());
               }
            });
            messages2.forEach(msg->{
                if(!msgSended.contains(msg.getUid())) {
                    kafkaTemplateBlogger.send(taskParam.getTopic(), msg);
                    msgSended.add(msg.getUid());
                }
            });
            XxlJobLogger.log("排行版播主类别={},投递消息数: params={}", b.getName().equals("ALL")?"":b.getName(),msgSended.size());
        });
    }
    public void collectBloggerMainFromBlogger(int bloggerLogType, String who,TaskParam taskParam){
        long start=System.currentTimeMillis();
        //获取位置
        BloggerLog bloggerLog=getBloggerLog(bloggerLogType,who);

        // 开始id
        long startId = bloggerLog.getCurrentBloggerId() + 1;
        // 结束的id
        long endId = startId + taskParam.getRecords();

        List<Blogger> pollBloggers = bloggerRepository.findSlimBloggersByIdAndCount(appConfig.getSysCollectFollowerCount(),startId,endId);

        //记录下一次发送的位置
        long id = endId;
        Integer pages = Integer.parseInt(bloggerLog.getExtValue()) + 1;
        if (CollectionUtils.isEmpty(pollBloggers)) {
            // 查询最大的id 判断是否到最后了
            long maxId = bloggerRepository.findMaxId();
            // 到最后了
            if(id >= maxId){
                id = 0;
                pages = 1;
                log.warn("startId={} ，无数据",bloggerLog.getCurrentBloggerId());
                XxlJobLogger.log("startId={} ，无数据",bloggerLog.getCurrentBloggerId());
            }
        }

        if (!CollectionUtils.isEmpty(pollBloggers)) {
            bloggerMainRepository.batchInsert(pollBloggers,who);
        }

        bloggerLog.setCurrentBloggerId(id);
        bloggerLog.setExtValue(pages.toString());
        bloggerLogRepository.save(bloggerLog);
        log.info(" who={}, size={},有效size={},startId={}, pageSize={},times={}秒",who,pollBloggers.size(),pollBloggers.size(),bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),(System.currentTimeMillis()-start)/1000);
        XxlJobLogger.log("  size={},有效size={},startId={}, pageSize={}, times={}秒",pollBloggers.size(),pollBloggers.size(),bloggerLog.getCurrentBloggerId(),taskParam.getRecords(),(System.currentTimeMillis()-start)/1000);


    }

    /**
     * 批量查询
     */
//    public List<BloggerMain> batchQueryBloggerMains(List<Long> ids){
//       int batchNums=500;
//        if(ids.size()<=batchNums){
//            return bloggerMainRepository.findByoUidIn(ids);
//        }
//
//        int cnt=0;
//        List<Long> tmpIds=new ArrayList<>(batchNums);
//        List<BloggerMain> rst=new ArrayList<>(ids.size());
//
//        for (Long e : ids) {
//            tmpIds.add(e);
//            cnt++;
//            if(cnt%batchNums==0){
//                rst.addAll(bloggerMainRepository.findByoUidIn(tmpIds)) ;
//                cnt=0;
//                tmpIds.clear();
//            }
//        }
//        //剩余
//        if(cnt>0){
//            rst.addAll(bloggerMainRepository.findByoUidIn(tmpIds)) ;
//            tmpIds.clear();
//        }
//        return rst;
//    }

}
